
#include "Connection.h"
#include <unistd.h>
#include "Buffer.h"
#include "Channel.h"
#include "Socket.h"
#include "macro.h"
#include "string.h"
#include "Log.h"

static Logger::ptr g_logger = MYWEB_LOG_ROOT();

Connection::Connection(int fd, EventLoop *loop) {
  m_socket = std::make_unique<Socket>();
  m_socket->set_fd(fd);
  if (loop != nullptr) {
    m_channel = std::make_unique<Channel>(fd, loop);
    m_channel->EnableRead();
    m_channel->EnableET();
  }
  m_read_buf = std::make_unique<Buffer>();
  m_send_buf = std::make_unique<Buffer>();

  m_state = State::Connected;
}

Connection::~Connection() {}

RC Connection::Read() {
  if(m_state != State::Connected) {
    MYWEB_LOG_ERROR(g_logger) << "Connection is not connected, can not read\n";
    return RC_CONNECTION_ERROR;
  }
  MYWEB_ASSERT(m_state == State::Connected && "connection state is disconnected!");
  m_read_buf->Clear();
  if (m_socket->IsNonBlocking()) {
    return ReadNonBlocking();
  } else {
    return ReadBlocking();
  }
}
RC Connection::Write() {
  if (m_state != State::Connected) {
    MYWEB_LOG_ERROR(g_logger) << "Connection is not connected, can not write\n";
    return RC_CONNECTION_ERROR;
  }
  RC rc = RC_UNDEFINED;
  if (m_socket->IsNonBlocking()) {
    rc = WriteNonBlocking();
  } else {
    rc = WriteBlocking();
  }
  m_send_buf->Clear();
  return rc;
}

RC Connection::ReadNonBlocking() {
  int sockfd = m_socket->fd();
  char buf[1024];  // 这个buf大小无所谓
  while (true) {   // 使用非阻塞IO，读取客户端buffer，一次读取buf大小数据，直到全部读取完毕
    memset(buf, 0, sizeof(buf));
    ssize_t bytes_read = read(sockfd, buf, sizeof(buf));
    if (bytes_read > 0) {
      m_read_buf->Append(buf, bytes_read);
    } else if (bytes_read == -1 && errno == EINTR) {  // 程序正常中断、继续读取
        MYWEB_LOG_INFO(g_logger) << "continue reading\n";
        continue;
    } else if (bytes_read == -1 &&
               ((errno == EAGAIN) || (errno == EWOULDBLOCK))) {  // 非阻塞IO，这个条件表示数据全部读取完毕
      break;
    } else if (bytes_read == 0) {  // EOF，客户端断开连接
        MYWEB_LOG_FMT_INFO(g_logger, "read EOF, client fd %d disconnected\n", sockfd);
        m_state = State::Closed;
        Close();
      break;
    } else {
        MYWEB_LOG_FMT_ERROR(g_logger, "Other error on client fd %d\n", sockfd);
        m_state = State::Closed;
        Close();
        break;
    }
  }
  return RC_SUCCESS;
}
RC Connection::WriteNonBlocking() {
  int sockfd = m_socket->fd();
  char buf[m_send_buf->Size()];
  memcpy(buf, m_send_buf->c_str(), m_send_buf->Size());
  int data_size = m_send_buf->Size();
  int data_left = data_size;
  while (data_left > 0) {
    ssize_t bytes_write = write(sockfd, buf + data_size - data_left, data_left);
    if (bytes_write == -1 && errno == EINTR) {
        MYWEB_LOG_INFO(g_logger) << "continue writing\n";
        continue;
    }
    if (bytes_write == -1 && errno == EAGAIN) {
      break;
    }
    if (bytes_write == -1) {
        MYWEB_LOG_FMT_ERROR(g_logger, "Other error on client fd %d\n", sockfd);
        m_state = State::Closed;
        break;
    }
    data_left -= bytes_write;
  }
  return RC_SUCCESS;
}

RC Connection::ReadBlocking() {
  int sockfd = m_socket->fd();
  char buf[1024];
  ssize_t bytes_read = read(sockfd, buf, sizeof(buf));
  if (bytes_read > 0) {
    m_read_buf->Append(buf, bytes_read);
  } else if (bytes_read == 0) {
        MYWEB_LOG_FMT_INFO(g_logger, "read EOF, client fd %d disconnected\n", sockfd);
        m_state = State::Closed;
  } else if (bytes_read == -1) {
    MYWEB_LOG_FMT_INFO(g_logger, "Other error on blocking client fd %d\n", sockfd);
    m_state = State::Closed;
  }
  return RC_SUCCESS;
}

RC Connection::WriteBlocking() {
  // 没有处理send_buffer_数据大于TCP写缓冲区，的情况，可能会有bug
  int sockfd = m_socket->fd();
  ssize_t bytes_write = write(sockfd, m_send_buf->buf().c_str(), m_send_buf->Size());
  if (bytes_write == -1) {
    MYWEB_LOG_FMT_INFO(g_logger, "Other error on blocking client fd %d\n", sockfd);
    m_state = State::Closed;
  }
  return RC_SUCCESS;
}

RC Connection::Send(std::string msg) {
  set_send_buf(msg.c_str());
  Write();
  return RC_SUCCESS;
}

void Connection::Business() {
  Read();
  m_on_recv_cb(this);
}

void Connection::set_delete_connection(std::function<void(int)> const &fn) { m_delete_connectioin_cb = fn; }

void Connection::set_on_recv(std::function<void(Connection *)> const &fn) {
  m_on_recv_cb = fn;
  m_channel->set_read_callback([this] {
      this->Business();
  });
}

void Connection::Close() { m_delete_connectioin_cb(m_socket->fd()); }

Connection::State Connection::state() const { return m_state; }

Socket *Connection::socket() const { return m_socket.get(); }

void Connection::set_send_buf(const char *str) { m_send_buf->set_buf(str); }
Buffer *Connection::read_buf() { return m_read_buf.get(); }
Buffer *Connection::send_buf() { return m_send_buf.get(); }
